package org.geektimes.reactive.message.streams;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

/**
 * @Author: jipeng
 * @Description:
 * @Date: Created in 2021/4/6 8:29
 */
public class DefaultSubscriber<T> implements Subscriber<T> {

    private Subscription subscription;
    private int count = 0;

    @Override
    public void onSubscribe(Subscription s) {
        this.subscription = s;
    }

    @Override
    public void onNext(T t) {
        System.out.println("收到数据 ：" + t);

        if(++count  == 3){ // 当到达阀值时，下次取消 Publisher 给当前 Subscriber 发送数据
            subscription.cancel();
            return;
        }
    }

    @Override
    public void onError(Throwable throwable) {
        System.out.println("遇到异常："+ throwable);
    }

    @Override
    public void onComplete() {
        System.out.println("收到数据完成");
    }
}
